package com.hliushi.spark.sql

import org.apache.spark.sql
import org.apache.spark.sql.{ColumnName, SparkSession}
import org.junit.Test

/**
 * descriptions:
 *
 * author: Hliushi
 * date: 2021/5/19 9:37
 */
class Column {


  // 1.创建spark对象
  private val spark: SparkSession = SparkSession.builder()
    .appName("column")
    .master("local[6]")
    .getOrCreate()

  import spark.implicits._

  /**
   * Column表示Dataset中的一个列, 并且可以持有一个表达式
   * 这个表达式作用于每一条数据, 对每条数据都生成一个值, 之所以有单独这样的一个章节
   * 是因为列的操作属于细节, 但是有比较常见, 会在很多算子中配合出现
   * #
   * column有几种创建方式, 一共六种 --> 四种无绑定, 两种有绑定
   * column对象可以用作于Dataset和DataFrame中
   * column可以和命令式的弱类型的API配合使用select where
   */
  @Test
  def createColumn(): Unit = {

    val ds = Seq(Person("zhangsan", 15), Person("lisi", 16), Person("wangwu", 20)).toDS()
    val ds1 = Seq(Person("zhangsan", 15), Person("lisi", 16), Person("wangwu", 20)).toDS()
    val df = Seq(("zhangsan", 15), ("lisi", 16), ("wangwu", 20)).toDF("name", "age")

    // 2. '  必须导入spark的隐式转换才能使用
    // final class Symbol private (val name: String) extends Serializable

    /**
     * SQLImplicits类中的 symbolToColumn方法
     * implicit def symbolToColumn(s: Symbol): ColumnName = new ColumnName(s.name)
     * .  所以这个 symbol 可以使用 Column类的方法
     * #
     * package org.apache.spark.sql  Column.scala
     * .      def asc: Column 方法
     */
    val symbol: Symbol = 'name
    val column: sql.Column = symbol

    // 3. $  必须导入spark的隐式转换才能使用
    val column1: ColumnName = $"name"

    // 4. col 必须导入functions
    import org.apache.spark.sql.functions._
    val column2: sql.Column = col("name")

    // 5. column 必须导入functions
    val column3 = column("name")

    /**
     * 这四种创建方式[无绑定ds], 有关联的Dataset吗?
     * -
     * Dataset可以使用Column, 那么DataFrame也可以使用Column对象选中列吗?
     * .    可以的
     * select方法可以使用column对象来选中某个列, 那么其他的算子也可以吗
     * .   def select(cols: Column*): DataFrame
     * .   def select(col: String, cols: String*): DataFrame
     * .   def selectExpr(exprs: String*): DataFrame
     * .   def where(condition: Column): Dataset[T]
     * .   def where(conditionExpr: String): Dataset[T]
     */
    ds.select(column).show()
    df.select(column).show()

    df.where(column.isNotNull).show()
    //  +--------+---+
    //  |    name|age|
    //  +--------+---+
    //  |zhangsan| 15|
    //  |    lisi| 16|
    //  |  wangwu| 20|
    //  +--------+---+

    /**
     * package org.apache.spark.sql  Column.scala
     * def === (other: Any): Column
     */
    df.where(column === "zhangsan").show()
    //  +--------+---+
    //  |    name|age|
    //  +--------+---+
    //  |zhangsan| 15|
    //  +--------+---+

    /**
     * 6. dataset.col
     * 使用Dataset来获取column对象, 会和某个Dataset进行绑定, 在逻辑计划中, 就会有不同的表现
     */
    val column4: sql.Column = ds.col("name")
    val column5: sql.Column = ds1.col("name")

    // resolved attribute(s) name#7 missing from name#2,age#3 in operator !Project [name#7];;
    // ds.select(column5).show()

    /**
     * 为什么要和Dataset来绑定呢?
     * 前面的Column对象创建方式所创建的Column对象都会Free[自由的]的,
     * 也就是没有绑定任何Dataset, 所有可以作用于任何Dataset
     * 同时也可以通过Dataset的col方法选择一个列, 但是这个Column是绑定的Dataset的,
     * 比如说column4和column5, 只能用于创建其的Dataset上
     * 所以 ds.select(column5).show()  --> 会报错了
     * column4绑定了ds, column5绑定了ds1, 所以可以使用 ds.join(ds1, column4 === column5)
     */
    ds.join(ds1, ds.col("name") === ds1.col("name")).show()
    ds.join(ds1, column4 === column5).show()
    //  +--------+---+--------+---+
    //  |    name|age|    name|age|
    //  +--------+---+--------+---+
    //  |zhangsan| 15|zhangsan| 15|
    //  |    lisi| 16|    lisi| 16|
    //  |  wangwu| 20|  wangwu| 20|
    //  +--------+---+--------+---+

    /**
     * 7. dataset.apply
     * 在伴生对象中
     * .    ds.apply("...") 其实就是 ds("...") 的别名
     */
    val column6: sql.Column = ds.apply("name")
    val column7: sql.Column = ds("name")
  }


  /**
   * as方法有两种用法, 通过as[Type]的形式可以将一个列中数据的类型换为Type类型
   * .               通过as(name)的形式使用as方法可以位列创建别名
   * #
   * def as(alias: String): Column
   * def as[U : Encoder]: TypedColumn[Any, U]
   */
  @Test
  def aliasAndTransform(): Unit = {
    val ds = Seq(Person("zhangsan", 15), Person("lisi", 16), Person("wangwu", 20)).toDS()

    // select name, count(age) as age_count from table group by name
    ds.select($"name".as("new_name")).show()
    //  +--------+
    //  |new_name|
    //  +--------+
    //  |zhangsan|
    //  |    lisi|
    //  |  wangwu|
    //  +--------+

    // 转换类型   把age的Int类型转为Long类型
    println("age类型准换之前!")
    ds.printSchema()
    val transform = ds.select($"age".as[Long])
    val transformString = ds.select($"age".as[String])
    println("age类型准换之后!")

    // x.getClass = long
    // x.getClass = long
    // x.getClass = long
    transform.foreach((x: Long) => {
      println(s"x.getClass = ${x.getClass}")
    })

    transform.printSchema()
    transformString.printSchema()

  }

  @Test
  def apiOp(): Unit = {
    val ds = Seq(Person("zhangsan", 15), Person("lisi", 16), Person("wangwu", 20)).toDS()

    // 需求一, ds增加列, 双倍年龄  Dataset.withColumn方法会调用Column.as()方法
    // $"age" * 2 其实本质上就是将一个表达式(逻辑计划表达式) 附加到Column对象上
    // 表达式在执行的时候对应每一条数据进行操作
    // 其中$"age" * 2   * 是Column的方法   def * (other: Any): Column
    // $"age" * 2 等同于 $"age".*(2)  这是scala的中缀调用方法, 是一个语法糖
    ds.withColumn("doubled01", $"age" * 2).show()
    ds.withColumn("doubled02", $"age".*(2)).show()

    // 需求二, 模糊查询
    // select * from table where name like zhang%
    ds.where($"name" like "zhang%").show()

    // 需求三, 排序, 正反序      其中sort就是orderBy的别名
    ds.sort($"age".asc).show()
    ds.orderBy($"age".desc).show()

    // 需求四, 枚举判断
    // select * from table name in("zhangsan", "wangwu", "zhaoliu")
    ds.where($"name" isin("zhangsan", "wangwu", "zhaoliu")).show()
  }
}